package com.zhanghe.study.mapreduce.join;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author zh
 * @date 2021/3/30 11:47
 */
public class JoinReducer extends Reducer<Text,TableBean,TableBean, NullWritable> {


    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
        List<TableBean> orderList = new ArrayList<>();
        TableBean productBean = new TableBean();
        // 因为相同的pid可能对应多个订单,而只会有一个对应的产品
        for(TableBean bean : values){
            // 订单
            if(StringUtils.equals(bean.getFlag(),"order")){
                TableBean order = new TableBean();
                order.setId(bean.getId());
                order.setPid(bean.getPid());
                order.setAmount(bean.getAmount());
                orderList.add(order);
            } else if(StringUtils.equals(bean.getFlag(),"product")){
                productBean.setName(bean.getName());
            }
        }

        for(TableBean tableBean : orderList){
            tableBean.setName(productBean.getName());
            context.write(tableBean,NullWritable.get());
        }
    }
}
